package com.zhu.gmall.flume.interceptor;


import com.zhu.gmall.flume.utils.JSONUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;

/**
 * 校验json日志数据是否完整 拦截器
 */
public class ETLInterceptor implements Interceptor {

    /**
     * 初始化
     */
    @Override
    public void initialize() {

    }

    /**
     * 单 event模式
     * @param event
     * @return
     */
    @Override
    public Event intercept(Event event) {
        byte[] body = event.getBody();   //body
        String log = new String(body, StandardCharsets.UTF_8);  //单条log

                //判断是不是合法json
        if(JSONUtil.isJSONValidate(log)){
            return event;
        }else {
            return null;
        }
    }

    /**
     * 多个 Event模式
     * @param list
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> list) {

        //index out Exception
        Iterator<Event> eventIterator = list.iterator();
        while (eventIterator.hasNext()){
            Event event = eventIterator.next();
            if(intercept(event) == null){
                eventIterator.remove();
            }
        }
        return list;
    }

    /**
     * 关闭资源
     */
    @Override
    public void close() {

    }

    public static class Builder implements  Interceptor.Builder{

        /**
         * 建造者模式 创建拦截器
         * @return
         */
        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }

        @Override
        public void configure(Context context) {
        }
    }
}
